package com.uam.core.dao.elasticsearch;

import java.io.BufferedReader;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.elasticsearch.action.admin.indices.alias.IndicesAliasesRequest;
import org.elasticsearch.action.admin.indices.alias.IndicesAliasesRequest.AliasActions;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.admin.indices.settings.put.UpdateSettingsRequest;
import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.get.MultiGetItemResponse;
import org.elasticsearch.action.get.MultiGetRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.support.WriteRequest.RefreshPolicy;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.client.IndicesClient;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.client.indices.GetIndexRequest;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.query.RangeQueryBuilder;
import org.elasticsearch.index.reindex.BulkByScrollResponse;
import org.elasticsearch.index.reindex.DeleteByQueryRequest;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.collapse.CollapseBuilder;
import org.elasticsearch.search.fetch.subphase.FetchSourceContext;
import org.elasticsearch.search.sort.SortBuilder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Repository;

import com.uam.core.entity.enums.ElasticsearchIndices;
import com.uam.core.entity.model.ElasticsearchQueryResult;
import com.uam.core.entity.model.ElasticsearchQueryResult.CommonFields;
import com.uam.core.entity.uitls.StringUtil;

import lombok.extern.slf4j.Slf4j;

@Repository
@Slf4j
public class ElasticsearchDao {

	public static final int[] OPERATION_FAILURE_STATUS = new int[] { -1, -1 };

	@Value("${charset}")
	private String charset = null;

	@Value("${spring.data.elasticsearch.rest.read-timeout}")
	private long requestTimeout = 0L;

	@Value("${elasticsearch.bulk.timeout}")
	private long bulkTimeout = 0L;

	@Value("${elasticsearch.index.replicas}")
	private int indexReplicas = 0;

	@Autowired
	private RestHighLevelClient elasticsearchClient = null;

	public long deleteDocumentByName( String index,String key,Object value) throws IOException {
		// 构建删除请求，删除具有特定 name 值的文档
		DeleteByQueryRequest deleteByQueryRequest = new DeleteByQueryRequest(index);
		deleteByQueryRequest.setQuery(QueryBuilders.termQuery(key, value));

		// 执行删除请求
		BulkByScrollResponse bulk=elasticsearchClient.deleteByQuery(deleteByQueryRequest, RequestOptions.DEFAULT);
		return bulk.getDeleted();
	}

	public int[] persistence(BulkRequest datas) {
		try {
			int i = 0;
			BulkItemResponse[] responses = this.elasticsearchClient.bulk(datas, RequestOptions.DEFAULT).getItems();
			int length = responses.length;
			int succeed = length;
			int failed = 0;
			while (i < length) {
				if (responses[i++].isFailed()) {
					--succeed;
					++failed;
				}
			}
			return new int[] { succeed, failed };
		} catch (Throwable e) {
			ElasticsearchDao.log.error(e.getMessage(), e);
		}
		return ElasticsearchDao.OPERATION_FAILURE_STATUS;
	}

	public boolean save(ElasticsearchIndices index, Map<String, Object> data) {
		return this.save(index, null, data, true);
	}

	public boolean save(ElasticsearchIndices index, Map<String, Object> data, boolean isImmediate) {
		return this.save(index, null, data, isImmediate);
	}

	public boolean save(ElasticsearchIndices index, String id, Map<String, Object> data) {
		return this.save(index, id, data, true);
	}

	public boolean save(ElasticsearchIndices index, String id, Map<String, Object> data, boolean isImmediate) {
		try {
			return this.elasticsearchClient
					.index(this.buildIndexRequest(index, id, data, isImmediate), RequestOptions.DEFAULT)
					.status() == RestStatus.OK;
		} catch (Throwable e) {
			ElasticsearchDao.log.error(e.getMessage(), e);
		}
		return false;
	}

	public boolean update(ElasticsearchIndices index, String id, Map<String, Object> data) {
		try {
			UpdateRequest request = new UpdateRequest(index.getName(), id);
			request.timeout(this.getRequestTimeout());
			request.setRefreshPolicy(RefreshPolicy.IMMEDIATE);
			request.doc(data, XContentType.JSON);
			//log.info("update:{},id:{},query:{}", index.getName(),id, request.toString());
			return this.elasticsearchClient.update(request, RequestOptions.DEFAULT).status() == RestStatus.OK;
		} catch (Throwable e) {
			ElasticsearchDao.log.error(e.getMessage(), e);
		}
		return false;
	}

	public boolean delete(ElasticsearchIndices index, String id) {
		try {
			DeleteRequest request = new DeleteRequest(index.getName(), id);
			request.timeout(this.getRequestTimeout());
			request.setRefreshPolicy(RefreshPolicy.IMMEDIATE);
			return this.elasticsearchClient.delete(request, RequestOptions.DEFAULT).status() == RestStatus.OK;
		} catch (Throwable e) {
			ElasticsearchDao.log.error(e.getMessage(), e);
		}
		return false;
	}

	public boolean exists(ElasticsearchIndices index, String id) {
		GetRequest request = new GetRequest(index.getName(), id);
		request.fetchSourceContext(new FetchSourceContext(false));
		request.storedFields("_none_");
		try {
			return this.elasticsearchClient.exists(request, RequestOptions.DEFAULT);
		} catch (IOException e) {
			ElasticsearchDao.log.error(e.getMessage(), e);
		}
		return false;
	}

	public ElasticsearchQueryResult detail(ElasticsearchIndices index, String id) {
		try {
			GetResponse response = this.elasticsearchClient.get(new GetRequest(index.getName(), id),
					RequestOptions.DEFAULT);
			if (response == null) {
				return ElasticsearchQueryResult.EMPTY;
			}
			Map<String, Object> data = response.getSourceAsMap();
			if (data != null && data.size() != 0) {
				data.put(CommonFields.ID.getName(), id);
				List<Map<String, Object>> datas = new ArrayList<Map<String, Object>>(1);
				datas.add(data);
				return new ElasticsearchQueryResult(1, datas);
			}
		} catch (IOException e) {
			ElasticsearchDao.log.error(e.getMessage(), e);
		}
		return ElasticsearchQueryResult.EMPTY;
	}

	public ElasticsearchQueryResult detail(ElasticsearchIndices index, String[] ids) {
		int i = 0;
		int size = ids.length;
		String indexName = index.getName();
		MultiGetRequest request = new MultiGetRequest();
		while (i < size) {
			request.add(indexName, ids[i++]);
		}
		try {
			MultiGetItemResponse[] responses = this.elasticsearchClient.mget(request, RequestOptions.DEFAULT)
					.getResponses();
			if (responses != null) {
				i = 0;
				size = responses.length;
				int total = 0;
				List<Map<String, Object>> datas = new ArrayList<Map<String, Object>>();
				while (i < size) {
					MultiGetItemResponse item = responses[i++];
					if (item == null) {
						continue;
					}
					GetResponse response = item.getResponse();
					if (response == null) {
						continue;
					}
					Map<String, Object> data = response.getSourceAsMap();
					if (data == null || data.size() == 0) {
						continue;
					}
					++total;
					data.put(CommonFields.ID.getName(), item.getId());
					datas.add(data);
				}
				return new ElasticsearchQueryResult(total, datas);
			}
		} catch (IOException e) {
			ElasticsearchDao.log.error(e.getMessage(), e);
		}
		return ElasticsearchQueryResult.EMPTY;
	}

	public ElasticsearchQueryResult query(QueryBuilder query, int pageNo, int pageSize, String[] includes,
			String[] excludes, ElasticsearchIndices index) {
		return this.query(query, pageNo, pageSize, null, includes, excludes, null, index);
	}

	public ElasticsearchQueryResult query(QueryBuilder query, int pageNo, int pageSize, SortBuilder<?> sort,
			String[] includes, String[] excludes, ElasticsearchIndices index) {
		return this.query(query, pageNo, pageSize, sort, includes, excludes, null, index);
	}

	public ElasticsearchQueryResult query(QueryBuilder query, int pageNo, int pageSize, SortBuilder<?> sort,
			String[] includes, String[] excludes, CollapseBuilder collapse, ElasticsearchIndices index) {
		String indexName = index.getName();
		/* ElasticsearchDao.log.info("index:{}, query:{}", indexName, query); */
		SearchRequest searchRequest = new SearchRequest(indexName);
		SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
		searchSourceBuilder.trackTotalHits(true);
		searchSourceBuilder.query(query);
		searchSourceBuilder.from((pageNo - 1) * pageSize);
		searchSourceBuilder.size(pageSize);
		searchSourceBuilder.timeout(this.getRequestTimeout());
		searchSourceBuilder.fetchSource(includes, excludes);
		if (sort != null) {
			searchSourceBuilder.sort(sort);
		}
		if (collapse != null) {
			searchSourceBuilder.collapse(collapse);
		}
		searchRequest.source(searchSourceBuilder);
		searchSourceBuilder = null;
		try {
			//log.info("index:{},执行sql：{}", Arrays.toString(searchRequest.indices()),searchRequest.source().toString());
			SearchResponse response = this.elasticsearchClient.search(searchRequest, RequestOptions.DEFAULT);
			RestStatus status = response.status();
			Boolean terminatedEarly = response.isTerminatedEarly();
			if (RestStatus.OK == status && !response.isTimedOut()
					&& (terminatedEarly == null || !terminatedEarly.booleanValue())) {
				int i = 0;
				SearchHits searchHits = response.getHits();
				int total = (int) searchHits.getTotalHits().value;
				if (total > 0) {
					SearchHit[] searchHitArr = searchHits.getHits();
					int length = searchHitArr.length;
					List<Map<String, Object>> datas = new ArrayList<Map<String, Object>>(length);
					while (i < length) {
						SearchHit searchHit = searchHitArr[i++];
						Map<String, Object> data = searchHit.getSourceAsMap();
						data.put(CommonFields.ID.getName(), searchHit.getId());
						datas.add(data);
					}
					return new ElasticsearchQueryResult(length, datas);
				}
			}
			status = null;
			terminatedEarly = null;
			response = null;
		} catch (Throwable e) {
			ElasticsearchDao.log.error(e.getMessage(), e);
		}
		return ElasticsearchQueryResult.EMPTY;
	}

	public void setReplicas(ElasticsearchIndices index) throws IOException {
		if (this.indexReplicas >= 0) {
			this.setReplicas(index, this.indexReplicas);
		}
	}

	public void setReplicas(ElasticsearchIndices index, int indexReplicas) throws IOException {
		this.elasticsearchClient.indices().putSettings(new UpdateSettingsRequest(index.getName())
				.settings(Settings.builder().put("index.number_of_replicas", indexReplicas)), RequestOptions.DEFAULT);
	}

	public void deleteIndex(ElasticsearchIndices index) {
		String indexName = index.getName();
		IndicesClient indicesClient = this.elasticsearchClient.indices();
		try {
			if (indicesClient.exists(new GetIndexRequest(indexName), RequestOptions.DEFAULT)) {
				indicesClient.delete(new DeleteIndexRequest(indexName), RequestOptions.DEFAULT);
			}
		} catch (Throwable e) {
			ElasticsearchDao.log.error(e.getMessage(), e);
		}
	}

	public IndicesAliasesRequest generateAliasesRequest(ElasticsearchIndices index, String[] alias) {
		return new IndicesAliasesRequest().timeout(this.getRequestTimeout())
				.addAliasAction(AliasActions.add().index(index.getName()).aliases(alias));
	}

	public BulkRequest generateBulkRequest() {
		return this.generateBulkRequest(true);
	}

	public BulkRequest generateBulkRequest(boolean isImmediate) {
		return new BulkRequest().timeout(this.getBulkTimeout())
				.setRefreshPolicy(isImmediate ? RefreshPolicy.IMMEDIATE : RefreshPolicy.NONE);
	}

	public boolean createIndex(ElasticsearchIndices index) {
		return this.createIndex(index, true);
	}

	public boolean createIndex(ElasticsearchIndices index, boolean deleteExist) {
		return this.createIndex(index.getName(), index.getSettingsFileName(), index.getMappingFileName(), deleteExist);
	}

	public boolean createIndex(String index) {
		return this.createIndex(index, null, null, true);
	}

	public boolean createIndex(String index, boolean deleteExist) {
		return this.createIndex(index, null, null, deleteExist);
	}

	public boolean createIndex(ElasticsearchIndices index, String settingsPath, String mappingPath,
			boolean deleteExist) {
		return this.createIndex(index.getName(), settingsPath, mappingPath, deleteExist);
	}

	public boolean createIndex(String index, String settingsPath, String mappingPath, boolean deleteExist) {
		IndicesClient indicesClient = this.elasticsearchClient.indices();
		try {
			if (deleteExist) {
				if (indicesClient.exists(new GetIndexRequest(index), RequestOptions.DEFAULT)) {
					indicesClient.delete(new DeleteIndexRequest(index), RequestOptions.DEFAULT);
				}
				return this.createIndex(index, settingsPath, mappingPath, indicesClient);
			} else {
				if (!indicesClient.exists(new GetIndexRequest(index), RequestOptions.DEFAULT)) {
					return this.createIndex(index, settingsPath, mappingPath, indicesClient);
				}
			}
		} catch (Throwable e) {
			ElasticsearchDao.log.error(e.getMessage(), e);
		}
		return false;
	}

	public boolean indexExists(ElasticsearchIndices index) {
		try {
			return this.elasticsearchClient.indices().exists(new GetIndexRequest(index.getName()),
					RequestOptions.DEFAULT);
		} catch (Throwable e) {
			ElasticsearchDao.log.error(e.getMessage(), e);
		}
		return false;
	}

	private boolean createIndex(String index, String settingsPath, String mappingPath, IndicesClient indicesClient) {
		CreateIndexRequest request = new CreateIndexRequest(index);
		ClassLoader classLoader = ElasticsearchDao.class.getClassLoader();
		if (!StringUtil.isEmpty(settingsPath)) {
			request.settings(this.readFile(classLoader.getResourceAsStream(settingsPath), this.charset),
					XContentType.JSON);
		}
		if (!StringUtil.isEmpty(mappingPath)) {
			request.mapping(this.readFile(classLoader.getResourceAsStream(mappingPath), this.charset),
					XContentType.JSON);
		}
		classLoader = null;
		try {
			indicesClient.create(request, RequestOptions.DEFAULT);
		} catch (Throwable e) {
			ElasticsearchDao.log.error(e.getMessage(), e);
			return false;
		}
		return true;
	}

	private String readFile(InputStream srcIn, String charset) {
		StringBuilder content = new StringBuilder();
		try {
			BufferedReader in = new BufferedReader(new InputStreamReader(srcIn, charset));
			String line = in.readLine();
			while (line != null) {
				content.append(line).append('\n');
				line = in.readLine();
			}
			in.close();
			in = null;
		} catch (UnsupportedEncodingException e) {
			ElasticsearchDao.log.error(e.getMessage(), e);
		} catch (FileNotFoundException e) {
			ElasticsearchDao.log.error(e.getMessage(), e);
		} catch (IOException e) {
			ElasticsearchDao.log.error(e.getMessage(), e);
		}
		int length = content.length();
		if (length == 0) {
			return content.toString();
		}
		return content.substring(0, length - 1);
	}

	private TimeValue getRequestTimeout() {
		return new TimeValue(this.requestTimeout, TimeUnit.MILLISECONDS);
	}

	private TimeValue getBulkTimeout() {
		return new TimeValue(this.bulkTimeout, TimeUnit.MILLISECONDS);
	}

	public IndexRequest buildIndexRequest(ElasticsearchIndices index, Map<String, Object> data) {
		return this.buildIndexRequest(index, null, data, true);
	}

	public IndexRequest buildIndexRequest(ElasticsearchIndices index, Map<String, Object> data, boolean isImmediate) {
		return this.buildIndexRequest(index, null, data, isImmediate);
	}

	public IndexRequest buildIndexRequest(ElasticsearchIndices index, String id, Map<String, Object> data) {
		return this.buildIndexRequest(index, id, data, true);
	}

	public IndexRequest buildIndexRequest(ElasticsearchIndices index, String id, Map<String, Object> data,
			boolean isImmediate) {
		IndexRequest indexRequest = new IndexRequest(index.getName()).timeout(this.getRequestTimeout()).source(data);
		if (!StringUtil.isEmpty(id)) {
			indexRequest.id(id);
		}
		if (isImmediate) {
			indexRequest.setRefreshPolicy(RefreshPolicy.IMMEDIATE);
		}
		return indexRequest;
	}

	public long count(QueryBuilder query, ElasticsearchIndices index) {
		String indexName = index.getName();
		/* ElasticsearchDao.log.info("index:{}, query:{}", indexName, query); */
		SearchRequest searchRequest = new SearchRequest(indexName);
		SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
		searchSourceBuilder.trackTotalHits(true);
		searchSourceBuilder.query(query);
		searchSourceBuilder.from(0);
		searchSourceBuilder.size(0);
		searchSourceBuilder.timeout(this.getRequestTimeout());
		searchSourceBuilder.fetchSource(false);
		searchRequest.source(searchSourceBuilder);
		searchSourceBuilder = null;
		try {
			return this.elasticsearchClient.search(searchRequest, RequestOptions.DEFAULT).getHits()
					.getTotalHits().value;
		} catch (Throwable e) {
			ElasticsearchDao.log.error(e.getMessage(), e);
		}
		return 0L;
	}

	/**
	 * 把删除的文档还原
	 *
	 * @param indexName   索引名称
	 * @param data 文档实体
	 * @return 响应
	 */
	public int updatePropJsonById(String indexName, List<Map<String,Object>> data) throws Exception {
		BulkRequest bulkRequest = new BulkRequest();
		data.forEach(doc -> {
			String id=doc.get("id")+"";
			doc.remove(id);
			UpdateRequest updateRequest = new UpdateRequest(indexName, id);
			String json = JSON.toJSONString(doc);
			updateRequest.doc(json, XContentType.JSON);
			bulkRequest.add(updateRequest);
		});
		bulkRequest.setRefreshPolicy(RefreshPolicy.IMMEDIATE);
		BulkResponse bulkResponse =this.elasticsearchClient.bulk(bulkRequest, RequestOptions.DEFAULT);
		// 检查是否有错误发生
		if (bulkResponse.hasFailures()) {
			ElasticsearchDao.log.error("更新文档失败:{}", bulkResponse.buildFailureMessage());
			throw new Exception("更新文档失败");
		}
		return bulkResponse.getItems().length;
	}

	public int saveBatchData(String indexName, List<Map<String,Object>> data) throws Exception {
		BulkRequest bulkRequest = new BulkRequest();
		data.forEach(doc -> {
			String id=doc.get("id")+"";
			doc.remove(id);
			IndexRequest updateRequest = new IndexRequest(indexName).id(id).source(doc);
			bulkRequest.add(updateRequest);
		});
		bulkRequest.setRefreshPolicy(RefreshPolicy.IMMEDIATE);
		BulkResponse bulkResponse =this.elasticsearchClient.bulk(bulkRequest, RequestOptions.DEFAULT);
		if (bulkResponse.hasFailures()) {
			ElasticsearchDao.log.error("添加文档失败:{}", bulkResponse.buildFailureMessage());
			throw new Exception("添加文档失败");
		}
		return bulkResponse.getItems().length;
	}

	/**
	 * @param indexName
	 * @param documentObjs
	 * @return
	 */
	public int batchDel(String indexName, List<String> documentObjs) {
		try {
			if (documentObjs == null) {
				return 0;
			}
			BulkRequest bulkRequest = new BulkRequest();
			documentObjs.forEach(doc -> {
				DeleteRequest deleteRequest = new DeleteRequest(indexName, doc);
				bulkRequest.add(deleteRequest);
			});
			bulkRequest.setRefreshPolicy(RefreshPolicy.IMMEDIATE.getValue());
			BulkResponse responses=this.elasticsearchClient.bulk(bulkRequest, RequestOptions.DEFAULT);
			return responses.getItems().length;
		} catch (Exception e) {
			log.error("batchDel:{}", e);
			e.printStackTrace();
		}
		return 0;
	}

	/**
	 * @param indexName
	 * @return
	 */
	public long delByQuery(String indexName, Map<String,Object> delWhere,
						String rangKey, Object rangGteValue,
						 Object rangLteValue) {
		try {
			if (delWhere == null) {
				return 0;
			}
			BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();

			if(rangKey !=null && rangGteValue !=null ) {
				RangeQueryBuilder queryBuilders=	QueryBuilders.rangeQuery(rangKey).gte(rangGteValue);
				boolQuery.must().add(queryBuilders);
			}

			if(rangKey !=null && rangLteValue !=null ) {
				RangeQueryBuilder queryBuilders=	QueryBuilders.rangeQuery(rangKey).lte(rangLteValue);
				boolQuery.must().add(queryBuilders);
			}

			if(delWhere !=null){
				delWhere.forEach((key,value)-> {
					boolQuery.must().add(QueryBuilders.matchQuery(key, value));
				});
			}
			DeleteByQueryRequest request = new DeleteByQueryRequest(indexName);
			request.setQuery(boolQuery);
			//System.out.println("执行查询删除："+boolQuery.toString());
			BulkByScrollResponse responses=this.elasticsearchClient.deleteByQuery(request, RequestOptions.DEFAULT);
			//System.out.println("删除数量："+responses.getDeleted());
			return responses.getDeleted();
		} catch (Exception e) {
			log.error("batchDel:{}", e);
			e.printStackTrace();
		}
		return 0;
	}

}
